#include "server.h"

static int _conn(struct listener_ctl *listener_ctl,
                struct ep_active_slot *active_slot)
{/*{{{*/
#define _MAX_NUM    9999
    SOCKET new_sock=active_slot->_u._conn.sock;
    int worker_idx;
    struct _worker_info *worker_info;
    int idle_num;
    int idle[WORKER_NUM];
    int min_busy_client_num;
    int min_busy_idx;
    int client_num;
    char msg[50];
    int msg_size;
    int res;

    pthread_mutex_lock(&share_info.worker_info_lock);
    idle_num=0;
    min_busy_client_num=_MAX_NUM;
    min_busy_idx=-1;
    for(worker_idx=0; worker_idx<WORKER_NUM; worker_idx++)
    {
        worker_info=&share_info.worker_info[worker_idx];
        if(worker_info->client_num == 0)
        {   idle[idle_num++]=worker_idx;   }
        else
        {
            if(min_busy_client_num > worker_info->client_num)
            {
                min_busy_client_num=worker_info->client_num;
                min_busy_idx=worker_idx;
            }
        }
    }
    if(idle_num > 0) // 选取无连接的worker
    {
        worker_idx=idle[random_int(0, idle_num-1, NULL)];
        if(cnfg_info._errlog_.tracking)
            errlog("%s %s: choose idle worker(%d)",
                    __FILE__, __func__, worker_idx);
    }
    else if(min_busy_client_num < _MAX_NUM) // 选取连接最少的worker
    {   worker_idx=min_busy_idx;   }
    else
    {
        errlog("%s %s: worker full, BAD",
                __FILE__, __func__);
        pthread_mutex_unlock(&share_info.worker_info_lock);
        sock_close(new_sock);
        return(-1);
    }
    client_num=++share_info.worker_info[worker_idx].client_num;
    pthread_mutex_unlock(&share_info.worker_info_lock);

    CLR_BUF(msg);
    para_put_item(msg, "NEWSLOT");
    para_put_item_int_fixed(msg, new_sock, SHORT_NUMBER_SIZE);
    msg_size=strlen(msg);
    if((res=ep_write(listener_ctl->ep,
                    listener_ctl->worker_slot[worker_idx],
                    msg, &msg_size)) != 0)
    {
        errlog("%s: listener ep_write worker(%d) fail, res:%d, "
                "abort, impossible",
                __func__, worker_idx, res);
        sock_close(new_sock);
        return(-1);
    }

    errlog("%s: listener send new slot to worker(%d) client_num:%d",
            __func__, worker_idx, client_num);
    return(0);
}/*}}}*/

void thrd_listener(struct listener_ctl *listener_ctl)
{/*{{{*/
    int res;
    struct ep_active_slot *active_list;
    
    while(1)
    {
        if((res=ep_poll(listener_ctl->ep, &active_list)) != 0)
        {
            errlog("%s %s: ep_poll fail, ret:%d, impossible",
                    __FILE__, __func__, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_CONN:
                    _conn(listener_ctl, active_list);
                    break;
                case EP_SLOT_ABORT:
                    errlog("%s: revoke slot(sock:%d,idx:%d) error(%d), "
                            "impossible",
                            __func__, active_list->_u._abort.sock,
                            active_list->slot_idx,
                            active_list->_u._abort.error);
                    break;
                default:
                    errlog("%s: listener poll type(%s) invalid, "
                            "impossible",
                            __func__, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }
    }
}/*}}}*/
